package paas.storage.distributedColumnDatabase.impl;

import com.alibaba.fastjson.JSON;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.RegionSplitter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import paas.storage.common.config.ConnectionManager;
import paas.storage.common.config.HBaseDaoUtil;
import paas.storage.common.constant.ResponseCode;
import paas.storage.common.utils.AssertUtils;
import paas.storage.distributedColumnDatabase.ITable;
import paas.storage.dto.ColumnFamilyModel;
import paas.storage.exception.RRException;
import paas.storage.result.*;

import java.io.IOException;
import java.util.*;
import java.util.regex.Pattern;

/**
 * 创建表实现类
 *
 * @author xufeng
 * @version 1.0
 * @date 2021/1/23
 */
@Service
public class ITableImpl implements ITable {

    private static Logger log = LoggerFactory.getLogger(IManagementImpl.class);

    @Autowired
    HBaseDaoUtil hBaseDaoUtil;

    @Autowired
    private ConnectionManager connectionManager;

    /**
     * 创建表
     *
     * @param connectionId      数据库连接标识 必填 c128
     * @param database          数据库   不必填 若不传递，则默认为default。 c64
     * @param table             数据表 必填 c64
     * @param columnFamilyList  [{"Name":列族，"Version":版本号，
     *                          "Compressio"n: 压缩格式，"TimeToLive"=
     *                          数据保存时间，"BlockCache"=是否开启缓
     *                          存}，{…}]
     *                          支持的压缩格式有 SNAPPY,GZ,LZ4,LZO
     *                          等列簇对象等 json数组字符串
     * @param numRegions        分区数 不必填 n4
     * @param splitMethod       分区方法 支持：HexStringSplit等方法。 不必填 c128
     *                          HexStringSplit: rowkey是十六进制的字符串作为前缀的
     *                          DecimalStringSplit: rowkey是10进制数字字符串作为前缀的
     *                          UniformSplit: rowkey前缀完全随机
     * @param regionReplication 分区副本数量 整型数值 不必填 n3
     * @param expendParams      扩展参数 c4000 不必填 以json字符串格式定义：
     *                          {key1:value1, key2:value2,…}。
     * @return
     */
    @Override
    public TableCreateResponse create(String connectionId, String database, String table, String columnFamilyList, Integer numRegions, String splitMethod, Integer regionReplication, String expendParams) {
        TableCreateResponse tableCreateResponse = new TableCreateResponse();
        tableCreateResponse.setTaskStatus(ResponseCode.TASK_STATUS_OK);
        try {
            //验证开始
            if (columnFamilyList.length() == 0) {
                throw new RRException("列簇不存在");
            }

            TableExistResponse tableExistResponse = exist(connectionId, database, table);
            if (tableExistResponse.getResult() == 1) {
                throw new RRException("表已存在");
            }
            if (connectionManager.getConnection(connectionId) == null) {
                tableCreateResponse.setTaskStatus(0);
                tableCreateResponse.setErrorMsg("Please establish a connection first");
                return tableCreateResponse;
            }
            TableName tn = hBaseDaoUtil.getTable(database, table);

            Admin admin = connectionManager.getConnection(connectionId).getAdmin();
            //创建命名空间
            if (StringUtils.isNotEmpty(database)) {
                //创建数据库columnFamilyList
                hBaseDaoUtil.createNameSpace(connectionId, database);
            }

            //解析
            List<ColumnFamilyModel> familyModels = JSON.parseArray(columnFamilyList, ColumnFamilyModel.class);

            TableDescriptorBuilder tableBuilder = TableDescriptorBuilder.newBuilder(tn);
            if (null != regionReplication && regionReplication > 0) {
                // 新建一个表描述，指定拆分策略和最大 StoreFile Size
                tableBuilder = TableDescriptorBuilder.newBuilder(tn)
                        //分区副本数
                        .setRegionReplication(regionReplication);
                            /*.setRegionSplitPolicyClassName(SteppingSplitPolicy.class.getName())
                            .setMaxFileSize(1048576000)
                            .setValue("hbase.increasing.policy.initial.size", "134217728");*/
            }
            //列簇信息
            List<ColumnFamilyDescriptor> columnFamilyDescriptors = new ArrayList<>();
            familyModels.forEach(columnFamilyModel -> {
                //创建列簇描述器
                ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder
                        .newBuilder(Bytes.toBytes(columnFamilyModel.getName()))
                        //最大版本号
                        .setMaxVersions(columnFamilyModel.getVersion())
                        //压缩格式
                        .setCompressionType(Compression.Algorithm.valueOf(columnFamilyModel.getCompression()))
                        //是否开启缓存
                        .setBlockCacheEnabled(columnFamilyModel.getBlockCache())
                        .build();
                columnFamilyDescriptors.add(columnFamilyDescriptor);
            });

            tableBuilder.setColumnFamilies(columnFamilyDescriptors);

            //分区方法不等于空
            if (null != numRegions && numRegions > 0 && StringUtils.isNotEmpty(splitMethod)) {
                byte[][] splitKeys;

                RegionSplitter.SplitAlgorithm splitAlgo = null;
                if ("HexStringSplit".equals(splitMethod)) {
                    splitAlgo = new RegionSplitter.HexStringSplit();
                }
                if ("DecimalStringSplit".equals(splitMethod)) {
                    splitAlgo = new RegionSplitter.DecimalStringSplit();
                }
                if ("UniformSplit".equals(splitMethod)) {
                    splitAlgo = new RegionSplitter.UniformSplit();
                }
                if (null != splitAlgo) {
                    splitKeys = splitAlgo.split(numRegions);
                    admin.createTable(tableBuilder.build(), splitKeys);
                }

            } else {
                admin.createTable(tableBuilder.build());
            }
            admin.close();
        } catch (RRException e) {
            tableCreateResponse.setTaskStatus(ResponseCode.TASK_STATUS_ERROR);
            tableCreateResponse.setErrorCode(ResponseCode.BUSINESS_CODE);
            tableCreateResponse.setErrorMsg("业务异常," + e.getMessage());
        }catch (Exception e) {
            tableCreateResponse.setTaskStatus(0);
            tableCreateResponse.setErrorCode(ResponseCode.TABLE_CREATE_ERROR);
            tableCreateResponse.setErrorMsg(e.getMessage());
            log.error(e.getMessage(), e);
        }
        tableCreateResponse.setTable(table);
        return tableCreateResponse;
    }


    @Override
    public TableDeleteResponse delete(String connectionId, String database, String table) {
        TableDeleteResponse tableDeleteResponse = new TableDeleteResponse();
        tableDeleteResponse.setTaskStatus(ResponseCode.TASK_STATUS_OK);
        try {
            AssertUtils.isTrue(StringUtils.isEmpty(connectionId), "连接不能为空");
            if (database != null) {
                AssertUtils.charLengthLe(database, 64, "连接长度为128以下");
            }
            AssertUtils.isTrue(StringUtils.isEmpty(table), "表名不能为空");
            AssertUtils.charLengthLe(table, 64, "表名不能为空");


            HTableDescriptor[] hTableDescriptors = new HTableDescriptor[0];

            if (connectionManager.getConnection(connectionId) == null) {
                tableDeleteResponse.setTaskStatus(0);
                tableDeleteResponse.setErrorMsg("Please establish a connection first");
                return tableDeleteResponse;
            }
            TableName tn;
            Admin admin = connectionManager.getConnection(connectionId).getAdmin();
            try {
                //判断是否存在
                TableExistResponse tableExistResponse = exist(connectionId, database, table);
                if (tableExistResponse.getResult() == 0) {
                    List<String> objects = new ArrayList<>();
                    objects.add(table);
                    tableDeleteResponse.setTableList(objects.toArray(new String[0]));
                    return tableDeleteResponse;
                }
                //正则表达式表名会报错
                tn = hBaseDaoUtil.getTable(database, table);
                //禁用表
                admin.disableTable(tn);
                //删除表
                admin.deleteTable(tn);

                List<String> tableList = new ArrayList<>();

                if (hTableDescriptors.length > 0) {
                    Arrays.stream(hTableDescriptors).forEach(hTableDescriptor -> {
                        tableList.add(hTableDescriptor.getNameAsString());
                    });
                } else {
                    tableList.add(table);
                }
                tableDeleteResponse.setTableList(tableList.toArray(new String[0]));
                return tableDeleteResponse;
            } catch (IllegalArgumentException e) {
                //格式包含非法字符 那就是正则删除表 不做处理
            }

            //正则表描述器
            database = StringUtils.isBlank(database) ? "default" : database;
            Pattern pattern = Pattern.compile(database + ":" + table);
            admin.disableTables(pattern);
            //正则表达式删除表
            hTableDescriptors = admin.deleteTables(pattern);

            List<String> tableList = new ArrayList<>();

            if (hTableDescriptors.length > 0) {
                Arrays.stream(hTableDescriptors).forEach(hTableDescriptor -> {
                    tableList.add(hTableDescriptor.getNameAsString());
                });
            } else {
                tableList.add(table);
            }
            tableDeleteResponse.setTableList(tableList.toArray(new String[0]));
            admin.close();
        }catch (RRException e) {
            tableDeleteResponse.setTaskStatus(ResponseCode.TASK_STATUS_ERROR);
            tableDeleteResponse.setErrorCode(ResponseCode.BUSINESS_CODE);
            tableDeleteResponse.setErrorMsg("业务异常," + e.getMessage());
        } catch (Exception e) {
            tableDeleteResponse.setTaskStatus(0);
            tableDeleteResponse.setErrorCode(ResponseCode.TABLE_DELETE_ERROR);
            tableDeleteResponse.setErrorMsg("删除表失败," + e.getMessage());
            log.error(e.getMessage(), e);
        }


        return tableDeleteResponse;
    }


    @Override
    public TableUpdateResponse update(String connectionId, String database, String table, String columnFamilyList,
                                      int autoFlush, int numRegions, String splitMethod, int regionReplication, String expendParams) {
        TableUpdateResponse tableUpdateResponse = new TableUpdateResponse();
        tableUpdateResponse.setTaskStatus(ResponseCode.TASK_STATUS_OK);
        tableUpdateResponse.setTable(table);
        try {
            //验证开始
            if (columnFamilyList.length() == 0) {
                throw new RRException("列簇不存在");
            }

            TableExistResponse tableExistResponse = exist(connectionId, database, table);
            if (tableExistResponse.getResult() == 0) {
                throw new RRException("表不存在");
            }
            if (connectionManager.getConnection(connectionId) == null) {
                tableUpdateResponse.setTaskStatus(0);
                tableUpdateResponse.setErrorCode(ResponseCode.TASK_STATUS_ERROR_CONNECTION);
                tableUpdateResponse.setErrorMsg("Please establish a connection first");
                return tableUpdateResponse;
            }

            TableName tn = hBaseDaoUtil.getTable(database, table);

            Admin admin = connectionManager.getConnection(connectionId).getAdmin();
            //构建新的表描述信息
            TableDescriptorBuilder newtableBuilder = TableDescriptorBuilder.newBuilder(tn);
            if (regionReplication > 0) {
                // 新建一个表描述，指定拆分策略和最大 StoreFile Size
                newtableBuilder = TableDescriptorBuilder.newBuilder(tn)
                        //分区副本数
                        .setRegionReplication(regionReplication);

            }
            //解析
            List<ColumnFamilyModel> familyModels = JSON.parseArray(columnFamilyList, ColumnFamilyModel.class);
            //列簇信息
            List<ColumnFamilyDescriptor> columnFamilyDescriptors = new ArrayList<>();
            familyModels.forEach(columnFamilyModel -> {
                //创建列簇描述器
                ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder
                        .newBuilder(Bytes.toBytes(columnFamilyModel.getName()))
                        //最大版本号
                        .setMaxVersions(columnFamilyModel.getVersion())
                        //压缩格式
                        .setCompressionType(Compression.Algorithm.valueOf(columnFamilyModel.getCompression()))
                        //是否开启缓存
                        .setBlockCacheEnabled(columnFamilyModel.getBlockCache())
                        .build();

                columnFamilyDescriptors.add(columnFamilyDescriptor);
            });
            //修改列簇信息
            newtableBuilder.setColumnFamilies(columnFamilyDescriptors);
            admin.modifyTable(newtableBuilder.build());
           /* //修改列簇信息
            //newtableBuilder.setColumnFamilies(columnFamilyDescriptors);
            TableDescriptorBuilder finalNewtableBuilder = newtableBuilder;
            TableDescriptorBuilder finalNewtableBuilder1 = newtableBuilder;
            columnFamilyDescriptors.forEach(columnFamilyDescriptor -> {
                finalNewtableBuilder.modifyColumnFamily(columnFamilyDescriptor);
                //修改表
                try {
                    admin.modifyTable(finalNewtableBuilder1.build());
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });*/

            byte[][] splitKeys;
            //分区方法不等于空
            if (StringUtils.isNotEmpty(splitMethod)) {
                RegionSplitter.SplitAlgorithm splitAlgo = null;
                if ("HexStringSplit".equals(splitMethod)) {
                    splitAlgo = new RegionSplitter.HexStringSplit();
                }
                if ("DecimalStringSplit".equals(splitMethod)) {
                    splitAlgo = new RegionSplitter.DecimalStringSplit();
                }
                if ("UniformSplit".equals(splitMethod)) {
                    splitAlgo = new RegionSplitter.UniformSplit();
                }
                splitKeys = splitAlgo.split(numRegions);
                //分区 //
                Arrays.stream(splitKeys).forEach(bytes -> {
                    try {
                        admin.split(tn, bytes);
                    } catch (IOException e) {
                        throw new RRException("分区出错" + e.getMessage());
                    }

                });
            }

        }catch (RRException e) {
            tableUpdateResponse.setTaskStatus(ResponseCode.TASK_STATUS_ERROR);
            tableUpdateResponse.setErrorCode(ResponseCode.BUSINESS_CODE);
            tableUpdateResponse.setErrorMsg("业务异常," + e.getMessage());
        } catch (Exception e) {
            tableUpdateResponse.setTaskStatus(ResponseCode.TASK_STATUS_OK);
            tableUpdateResponse.setErrorCode(ResponseCode.TABLE_UPDATE_ERROR);
            tableUpdateResponse.setErrorMsg(e.getMessage());
        }

        return tableUpdateResponse;
    }

    /**
     * 查询所有表
     *
     * @param connectionId
     * @param database     不必填   命名空间 二者选其一
     * @param table        表名 不必填  二者选其一
     * @return
     */
    @Override
    public TableListResponse list(String connectionId, String database, String table) {
        //返回数据
        TableListResponse tableListResponse = new TableListResponse();
        tableListResponse.setTaskStatus(1);

        try {
            AssertUtils.isTrue(StringUtils.isEmpty(connectionId), "连接不能为空");
            if (database != null) {
                AssertUtils.charLengthLe(database, 64, "连接长度为128以下");
            }
            if (table != null) {
                AssertUtils.charLengthLe(table, 64, "表名不能为空");
            }

            if (connectionManager.getConnection(connectionId) == null) {
                tableListResponse.setTaskStatus(0);
                tableListResponse.setErrorCode(ResponseCode.TASK_STATUS_ERROR_CONNECTION);
                tableListResponse.setErrorMsg("Please establish a connection first");
                return tableListResponse;
            }

            //根据数据库查询所有表
            Admin admin = connectionManager.getConnection(connectionId).getAdmin();
            if (StringUtils.isNotEmpty(database) && StringUtils.isEmpty(table)) {
                TableName[] tableNames = admin.listTableNamesByNamespace(database);
                List<String> stringList = new ArrayList<>();
                for (TableName o : tableNames) {
                    stringList.add(o.getNameAsString());
                }
                String[] arr = stringList.toArray(new String[0]);
                tableListResponse.setTableList(arr);
            }
            if (StringUtils.isNotEmpty(table)) {
                //正则表描述器
                database = StringUtils.isBlank(database) ? "default:" : database;
                Pattern pattern = Pattern.compile(database + ":" + table);
                List<TableDescriptor> tableDescriptors1 = admin.listTableDescriptors(pattern);
                if (tableDescriptors1.size() > 0) {
                    List<String> stringList = new ArrayList<>();
                    tableDescriptors1.forEach(tableDescriptor -> {
                        stringList.add(tableDescriptor.getTableName().getNameAsString());
                    });
                    String[] arr = stringList.toArray(new String[0]);
                    tableListResponse.setTableList(arr);
                    return tableListResponse;
                }
                //这里只支持字母和数字
                TableName tableName = null;

                try {
                    //这里只支持字母和数字 //正则会报错IllegalArgumentException
                    tableName = hBaseDaoUtil.getTable(database, table);
                } catch (IllegalArgumentException e) {
                    tableListResponse.setTableList(new String[0]);
                    return tableListResponse;
                }
                //根据表明查询表
                List<TableName> names = new ArrayList<>();
                names.add(tableName);
                //非正则的时候
                List<TableDescriptor> tableDescriptors = admin.listTableDescriptors(names);
                List<String> stringList = new ArrayList<>();
                tableDescriptors.forEach(tableDescriptor -> {
                    stringList.add(tableDescriptor.getTableName().getNameAsString());
                });
                String[] arr = stringList.toArray(new String[0]);
                tableListResponse.setTableList(arr);

            }
            admin.close();
        } catch (IOException e) {
            tableListResponse.setTaskStatus(ResponseCode.TASK_STATUS_ERROR);
            tableListResponse.setErrorCode(ResponseCode.TABLE_LIST_ERROR);
            tableListResponse.setErrorMsg("查询所有表失败");
            log.error(e.getMessage(), e);
        }

        return tableListResponse;
    }

    /**
     * 查询表
     *
     * @param connectionId 连接id
     * @param database     数据库
     * @param tableList    支持以列表形式传递1个或多个数据  (多个用英文逗号隔开)
     * @return
     */
    @Override
    public TableQueryResponse query(String connectionId, String database, String tableList) {
        TableQueryResponse tableQueryResponse = new TableQueryResponse();
        tableQueryResponse.setTaskStatus(1);

        try {

            AssertUtils.isTrue(StringUtils.isEmpty(connectionId), "连接不能为空");
            if (database != null) {
                AssertUtils.charLengthLe(database, 64, "连接长度为128以下");
            }
            if (tableList != null) {
                AssertUtils.charLengthLe(tableList, 64, "表名不能为空");
            }


            if (connectionManager.getConnection(connectionId) == null) {
                tableQueryResponse.setTaskStatus(0);
                tableQueryResponse.setErrorCode(ResponseCode.TASK_STATUS_ERROR_CONNECTION);
                tableQueryResponse.setErrorMsg("Please establish a connection first");
                return tableQueryResponse;
            }

            Admin admin = connectionManager.getConnection(connectionId).getAdmin();
            String[] str = tableList.split(",");
            //根据表明查询表
            List<TableName> names = new ArrayList<>();
            Arrays.stream(str).forEach(s -> {
                names.add(hBaseDaoUtil.getTable(database, s));
            });
            //表描述器
            List<TableDescriptor> tableDescriptors = admin.listTableDescriptors(names);
            //返回结果
            List<Map<String, Object>> vo = new ArrayList<>();
            //遍历表描述器
            tableDescriptors.forEach(tableDescriptor -> {
                ColumnFamilyModel columnFamilyModel = new ColumnFamilyModel();
                columnFamilyModel.setName(tableDescriptor.getTableName().getNameAsString());
                //列簇对象
                Map<String, Object> tableObj = new HashMap<>();
                //组装列簇
                List<ColumnFamilyModel> columnFamilyModelList = new ArrayList<>();
                Arrays.stream(tableDescriptor.getColumnFamilies()).forEach(columnFamilyDescriptor -> {
                    columnFamilyModel.setVersion(columnFamilyDescriptor.getMaxVersions());
                    columnFamilyModel.setBlockCache(columnFamilyDescriptor.isBlockCacheEnabled());
                    columnFamilyModel.setCompression(columnFamilyDescriptor.getCompressionType().getName());
                    columnFamilyModel.setTimeToLive(columnFamilyDescriptor.getTimeToLive());
                    columnFamilyModelList.add(columnFamilyModel);
                });
                tableObj.put(tableDescriptor.getTableName().getNameAsString(), columnFamilyModelList);
                vo.add(tableObj);
            });
            tableQueryResponse.setTableInfo(JSON.toJSONString(vo));
            admin.close();
        } catch (IOException e) {
            tableQueryResponse.setTaskStatus(0);
            tableQueryResponse.setErrorCode(ResponseCode.TABLE_QUERY_ERROR);
            tableQueryResponse.setErrorMsg("查询表失败," + e.getMessage());
            log.error(e.getMessage(), e);
        }
        return tableQueryResponse;
    }

    @Override
    public TableExistResponse exist(String connectionId, String database, String table) {
        TableExistResponse tableExistResponse = new TableExistResponse();
        tableExistResponse.setTaskStatus(1);
        try {
            AssertUtils.isTrue(StringUtils.isEmpty(connectionId), "连接不能为空");
            if (database != null) {
                AssertUtils.charLengthLe(database, 64, "连接长度为128以下");
            }
            if (table != null) {
                AssertUtils.charLengthLe(table, 64, "表名不能为空");
            }

            if (connectionManager.getConnection(connectionId) == null) {
                tableExistResponse.setTaskStatus(0);
                tableExistResponse.setErrorCode(ResponseCode.TASK_STATUS_ERROR_CONNECTION);
                tableExistResponse.setErrorMsg("Please establish a connection first");
                return tableExistResponse;

            }
            Connection conn = connectionManager.getConnection(connectionId);
            TableName tn = hBaseDaoUtil.getTable(database, table);

            Admin admin = conn.getAdmin();
            //表是否存在
            boolean flag = admin.tableExists(tn);
            tableExistResponse.setResult(flag ? 1 : 0);
        }catch (RRException e) {
            tableExistResponse.setTaskStatus(ResponseCode.TASK_STATUS_ERROR);
            tableExistResponse.setErrorCode(ResponseCode.BUSINESS_CODE);
            tableExistResponse.setErrorMsg("业务异常," + e.getMessage());
        } catch (Exception e) {
            tableExistResponse.setTaskStatus(0);
            tableExistResponse.setErrorCode(ResponseCode.TABLE_EXIST_ERROR);
            tableExistResponse.setErrorMsg("查询表是否存在失败," + e.getMessage());
            log.error(e.getMessage(), e);

        }
        return tableExistResponse;
    }


}
